package com.shujia.flink.core

import org.apache.flink.api.common.state.{ListStateDescriptor, MapStateDescriptor}
import org.apache.flink.api.common.state.ListState
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object Demo9Broadcast {
  def main(args: Array[String]): Unit = {
    /**
      * 动态拦截黑名单
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val ds = env.socketTextStream("node1", 8888)

    val mapStateDescriptor = new MapStateDescriptor[String, String](
      "black",
      BasicTypeInfo.STRING_TYPE_INFO,
      BasicTypeInfo.STRING_TYPE_INFO
    )

    //广播流  在每一个task中都有数据
    val broadcastDS = ds.broadcast(mapStateDescriptor)


    val ds1 = env.socketTextStream("node1", 9999)

    ds1
      .connect(broadcastDS) // 连接两个流
      .process(new MyBroadcastProcessFunction())
      .print()

    env.execute()
  }
}

class MyBroadcastProcessFunction extends BroadcastProcessFunction[String, String, String] {
  var black: ListBuffer[String] = _

  override def open(parameters: Configuration): Unit = {
    //保存黑名单
    black = new ListBuffer[String]
  }

  // ds1中每一天数据会被调用一次
  override def processElement(value: String, ctx: BroadcastProcessFunction[String, String, String]#ReadOnlyContext, out: Collector[String]): Unit = {
    if (black.contains(value)) {
      out.collect("监控到数据出现：" + value)
    }


  }

  //中博六中每一天数据会调用一次
  override def processBroadcastElement(value: String, ctx: BroadcastProcessFunction[String, String, String]#Context, out: Collector[String]): Unit = {
    //增加到黑名单
    black += value
  }
}
